package org.redis.cluster.transport.netty.server;

import java.net.InetSocketAddress;

import org.redis.cluster.RedisCluster;
import org.redis.cluster.config.ServerConfig;
import org.redis.cluster.config.SystemConfig;
import org.redis.cluster.plugin.Plugin;
import org.redis.cluster.transport.exception.TransportException;
import org.redis.cluster.transport.netty.NettySystemConfigs;
import org.redis.cluster.transport.netty.processor.NettyHandler;
import org.redis.cluster.transport.netty.protocol.NettyDecoder;
import org.redis.cluster.transport.netty.protocol.NettyEncoder;
import org.redis.cluster.transport.netty.session.NettySessionContext;
import org.redis.cluster.transport.protocol.ProtocolDecoder;
import org.redis.cluster.transport.protocol.ProtocolEncoder;
import org.redis.cluster.transport.server.Server;
import org.redis.cluster.transport.server.ServerChannel;
import org.redis.cluster.transport.session.SessionContext;
import org.redis.cluster.transport.session.SessionProcessor;

import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.AdaptiveRecvByteBufAllocator;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelHandler;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelOption;
import io.netty.channel.ChannelPipeline;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioServerSocketChannel;

public class NettyServer implements Server,Plugin {
	
	private final static int MIN_BUFFER=64;
	private final static int MAX_BUFFER=65535;
	
	private SystemConfig systemConfig;
	private ServerConfig config;
	private ProtocolDecoder decoder;
	private ProtocolEncoder encoder;
	private SessionProcessor<?> processor;

	public NettyServer(ServerConfig config,ProtocolDecoder decoder,ProtocolEncoder encoder,SessionProcessor<?> processor) {
		this.config=config;
		this.decoder=decoder;
		this.encoder=encoder;
		this.processor=processor;
	}
	
	@Override
	public void init(RedisCluster redisCluster) {
		this.systemConfig=redisCluster.getConfigManager().getSystemConfig();
	}
	
	@Override
	public void destroy() {
	}

	@Override
	public ServerChannel startup() {
		EventLoopGroup bossGroup=new NioEventLoopGroup(systemConfig.get(NettySystemConfigs.BOSS_THREAD).getValue(Integer.class,NettySystemConfigs.DEFAULT_BOSS_THREAD));
		EventLoopGroup workerGroup=new NioEventLoopGroup(config.getThread());
        
    	ServerBootstrap serverBootstrap=new ServerBootstrap();
		serverBootstrap.group(bossGroup,workerGroup);
		serverBootstrap.channel(NioServerSocketChannel.class);
		
		serverBootstrap.childHandler(new ChannelInitializer<SocketChannel>() {
			@Override
			protected void initChannel(SocketChannel socketChannel) throws Exception {
				SessionContext context=new NettySessionContext();
				
				ChannelPipeline pipeline=socketChannel.pipeline();
				pipeline.addLast("decoder", getDecoder(context));  
		        pipeline.addLast("encoder", getEncoder(context));  
		        pipeline.addLast("handler", getHandler(context));
			}
		});
		
		doProcessServerBootstrapConfig(serverBootstrap);
		
		ChannelFuture channelFuture=null;
		try {
			channelFuture=serverBootstrap.bind(new InetSocketAddress(config.getPort())).sync();
		}
		catch(Exception e) {
			throw new TransportException(e);
		}
		
		return new NettyServerChannel(channelFuture,bossGroup,workerGroup);
	}
	
	protected ChannelHandler getDecoder(SessionContext context) {
		return new NettyDecoder(decoder,context);
	}
	
	protected ChannelHandler getEncoder(SessionContext context) {
		return new NettyEncoder(encoder);
	}
	
	protected ChannelHandler getHandler(SessionContext context) {
		return new NettyHandler(processor,context);
	}
	
	protected void doProcessServerBootstrapConfig(ServerBootstrap serverBootstrap) {
		serverBootstrap.option(ChannelOption.CONNECT_TIMEOUT_MILLIS, config.getTimeout());
		serverBootstrap.childOption(ChannelOption.CONNECT_TIMEOUT_MILLIS, config.getTimeout());
		
		serverBootstrap.option(ChannelOption.SO_KEEPALIVE, true);
		serverBootstrap.childOption(ChannelOption.SO_KEEPALIVE, true);
		
		serverBootstrap.option(ChannelOption.TCP_NODELAY, true);
		serverBootstrap.childOption(ChannelOption.TCP_NODELAY, true);
		
		serverBootstrap.option(ChannelOption.RCVBUF_ALLOCATOR, new AdaptiveRecvByteBufAllocator(MIN_BUFFER,config.getBufferSize(),MAX_BUFFER));
		serverBootstrap.childOption(ChannelOption.RCVBUF_ALLOCATOR, new AdaptiveRecvByteBufAllocator(MIN_BUFFER,config.getBufferSize(),MAX_BUFFER));
	}
}